package com.magenta.maxoptra.gis.distancematrix;

import com.magenta.maxoptra.gis.commons.Logger;
import com.magenta.maxoptra.gis.distancematrix.events.EventsFactory;
import org.jetbrains.annotations.Nullable;

import javax.jms.*;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

/**
 * @author Sergey Grachev
 */
final class EventsProducer {

    private final Logger logger = new Logger(EventsProducer.class);
    private final Map<String, QueueSender> senders = new HashMap<String, QueueSender>();
    private final Map<String, Subscriber> subscribers = new HashMap<String, Subscriber>();

    private InitialContext ctx;
    private QueueConnectionFactory factory;
    private QueueConnection connection;
    private QueueSession session;

    @SuppressWarnings("unchecked")
    private <T> T lookup(final String name) throws NamingException {
        if (ctx == null) {
            ctx = new InitialContext();
        }
        return (T) ctx.lookup(name);
    }

    @Nullable
    private QueueSender getQueueSender(final String queueName) {
        synchronized (senders) {

            // initialize
            if (factory == null) {
                try {
                    factory = lookup("java:/ConnectionFactory");
                    connection = factory.createQueueConnection();
                    session = connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
                } catch (Exception e) {
                    logger.error("Can't create queue session", e);
                    return null;
                }
            }

            QueueSender sender = senders.get(queueName);
            if (sender == null) {
                try {
                    final Queue queue = lookup(queueName);
                    sender = session.createSender(queue);
                } catch (Exception e) {
                    logger.error("Can't create sender for queue = " + queueName, e);
                    return null;
                }
                senders.put(queueName, sender);
            }

            return sender;
        }
    }

    public void sendEvent(final Event event) {
        synchronized (subscribers) {
            for (final Map.Entry<String, Subscriber> entry : subscribers.entrySet()) {
                final Subscriber subscriber = entry.getValue();
                if (!subscriber.isSubscribedOn(event)) {
                    continue;
                }

                final QueueSender sender = getQueueSender(entry.getKey());
                if (sender == null) {
                    continue;
                }

                try {
                    sender.send(session.createObjectMessage(event));
                } catch (Exception e) {
                    logger.error("Can't send event to subscriber[" + entry.getKey() + "]", e);
                }
            }
        }
    }

    public void close() {
        synchronized (senders) {
            try {
                connection.close();
            } catch (JMSException e) {
                logger.error(e);
            }
            factory = null;
            connection = null;
            session = null;
            senders.clear();
            subscribers.clear();
        }
    }

    public void subscribe(final String queueName, final Set<EventType> events) {
        synchronized (subscribers) {
            Subscriber subscriber = subscribers.get(queueName);
            if (subscriber == null) {
                subscribers.put(queueName, subscriber = new Subscriber());
            }
            subscriber.addEvents(events);
        }
    }

    public void unsubscribe(final String queueName, final Set<EventType> events) {
        synchronized (subscribers) {
            final Subscriber subscriber = subscribers.get(queueName);
            if (subscriber == null) {
                return;
            }
            subscriber.removeEvents(events);
            if (subscriber.hasNoEvents()) {
                subscribers.remove(queueName);
            }
        }
    }

    public void notifyMatrixUpdateProgress(final long groupId, final long matrixId, final long total, final long progress) {
        sendEvent(EventsFactory.newMatrixUpdateProgress(groupId, matrixId, total, progress));
    }

    public void notifyMatrixUpdateFinished(final long groupId, final long matrixId, final UpdateResultType resultType) {
        sendEvent(EventsFactory.newMatrixUpdateFinished(groupId, matrixId, resultType));
    }

    private final static class Subscriber {
        private final Set<EventType> events = new HashSet<EventType>(0);

        public void addEvents(final Set<EventType> events) {
            this.events.addAll(events);
        }

        public void removeEvents(final Set<EventType> events) {
            this.events.removeAll(events);
        }

        public boolean hasNoEvents() {
            return events.isEmpty();
        }

        public boolean isSubscribedOn(final Event event) {
            return events.contains(event.getType());
        }
    }
}
